package com.xiguaji.tiktok.task.market;

import com.gugee.tiktok.common.model.dal.market.BloggerMarketRepository;
import com.gugee.tiktok.data.common.msg.MarketMessage;
import com.xiguaji.boot.toolkit.JsonUtil;
import com.xiguaji.tiktok.task.common.config.AppConfig;
import com.xiguaji.tiktok.task.common.job.TaskParam;
import com.xiguaji.tiktok.task.market.service.MarkerForkJoinTask;
import com.xiguaji.tiktok.task.market.spider.MarketSpiderClient;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;

@Component
@Slf4j
public class MarketTask {

    @Autowired
    AppConfig appConfig;
    @Autowired
    @Qualifier("kafkaTemplateMarket")
    KafkaTemplate<Long, MarketMessage> kafkaTemplateMarket;
    @Autowired
    MarketSpiderClient marketSpiderClient;
    @Autowired
    BloggerMarketRepository bloggerMarketRepository;
    @Autowired
    RestTemplate restTemplate;
//    @XxlJob("fetchBloggerMarketHandler")
//    public ReturnT<String> fetchBloggerMarketHandler(String param) {
//        val fetchBloggerMarketEvent = new FetchBloggerMarketEventListener.FetchBloggerMarketEvent(false);
//        eventBusTemplate.post(fetchBloggerMarketEvent);
//        return ReturnT.SUCCESS;
//    }
//    @XxlJob("fetchLoopBloggerMarketHandler")
//    public ReturnT<String> fetchLoopBloggerMarketHandler(String param) {
//        val fetchBloggerMarketEvent = new FetchBloggerMarketEventListener.FetchBloggerMarketEvent(false);
//        eventBusTemplate.post(fetchBloggerMarketEvent);
//        return ReturnT.SUCCESS;
//    }
//    @XxlJob("updataMarketHandler")
//    public ReturnT<String>  updataMarketHandler(String param){
//        long start = System.currentTimeMillis();
//        try {
//            log.info("星图数据更新任务: params={}", param);
//            XxlJobLogger.log("星图数据更新任务任务: params={}", param);
//            TaskParam jobParam = JsonUtil.jsonParse(TaskParam.class, param);
//            int bloggerLogType = BloggerLog.BloggerLogType.MARKER_UPDATA.getValue();
//           int code= markerService.updateMarkerHande(bloggerLogType,"updataMarketHandler",jobParam);
//            if(code==0){
//                return ReturnT.SUCCESS;
//            }
//            return ReturnT.FAIL;
//        } catch (Exception ex) {
//            XxlJobLogger.log(ex);
//            log.error(ex.getMessage());
//            return ReturnT.FAIL;
//        }finally {
//            log.info("结束  end cost total={},", (System.currentTimeMillis() - start) / 1000);
//            XxlJobLogger.log("结束  end cost total={}, params={}", (System.currentTimeMillis() - start) / 1000,param);
//        }
//    }
    @XxlJob("updateMarketByFansRangeByHandler")
    public ReturnT<String>  updateMarketByFansRangeByHandler(String param){
        long start = System.currentTimeMillis();
        Integer result =0;

        try {
            log.info("星图数据限制采集页数方式的更新任务: params={}", param);
            XxlJobLogger.log("星图数据限制采集页数方式的更新任务: params={}", param);
            TaskParam jobParam = JsonUtil.jsonParse(TaskParam.class, param);
            ForkJoinPool pool = new ForkJoinPool();


            MarkerForkJoinTask.MarketForkJoinParam params=new MarkerForkJoinTask.MarketForkJoinParam(1,200000000,jobParam.getPagesLimit()*jobParam.getRecords(),jobParam.getTopic());
//            if(marketSpiderClient==null || restTemplate==null ||kafkaTemplateMarket==null ||appConfig==null ){
//                log.error("appConfig={},params={},kafkaTemplateMarket={}", appConfig,params, kafkaTemplateMarket);
//                return ReturnT.FAIL;
//            }
            MarkerForkJoinTask task= new MarkerForkJoinTask(params,marketSpiderClient,restTemplate,kafkaTemplateMarket,appConfig);
            task.init();

            ForkJoinTask<Integer> taskFuture= pool.submit(task);
              result = taskFuture.get();
            return ReturnT.SUCCESS;

        } catch (Exception ex) {
            XxlJobLogger.log(ex);
            log.error(ex.getMessage());
            return ReturnT.FAIL;
        }finally {
            log.info("结束  end cost total={},采集数={}", (System.currentTimeMillis() - start) / 1000,result);
            XxlJobLogger.log("结束  end cost total={}, params={},采集数={}", (System.currentTimeMillis() - start) / 1000,param,result);
        }
    }
}
